package com.isyscore.os.metadata.kettle.step;

import com.isyscore.os.metadata.enums.KettleDataFlowNodeType;
import com.isyscore.os.metadata.kettle.base.*;
import com.isyscore.os.metadata.kettle.vis.SortField;
import com.isyscore.os.metadata.kettle.vis.SortRowsNode;
import com.isyscore.os.metadata.kettle.vis.UniqueNode;
import com.isyscore.os.metadata.kettle.vis.VisNode;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.plugins.PluginInterface;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;
import org.pentaho.di.trans.step.errorhandling.StreamInterface;
import org.pentaho.di.trans.steps.filterrows.FilterRowsMeta;
import org.pentaho.di.trans.steps.uniquerows.UniqueRowsMeta;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class UniqueStep extends AbstractStep {
    @Override
    public StepMeta decode(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws Exception {
        UniqueNode uniqueNode =(UniqueNode)step;
        PluginRegistry registry = PluginRegistry.getInstance();
        PluginInterface sp = registry.findPluginWithId( StepPluginType.class,  KettleDataFlowNodeType.Unique.name() );
        StepMetaInterface stepMetaInterface = (StepMetaInterface) registry.loadClass( sp );
        UniqueRowsMeta uniqueRowsMeta = (UniqueRowsMeta) stepMetaInterface;
        uniqueRowsMeta.setCompareFields(uniqueNode.getUniqueFields().toArray(new String[0]));
        //大小写敏感，默认否
        boolean[] caseInsensitive= new boolean[uniqueNode.getUniqueFields().size()];

        for (int i = 0; i < uniqueNode.getCaseInsensitives().size(); i++) {
            caseInsensitive[i] = uniqueNode.getCaseInsensitives().get(i);
        }

//        Arrays.fill(caseInsensitive, Boolean.FALSE);

        uniqueRowsMeta.setCaseInsensitive(caseInsensitive);
        uniqueRowsMeta.setCountRows(Boolean.FALSE);
        uniqueRowsMeta.setRejectDuplicateRow(Boolean.FALSE);
        StepMeta stepMeta = new StepMeta(KettleDataFlowNodeType.Unique.name() , uniqueNode.getNodeName(), stepMetaInterface);
        stepMeta.setDraw(true);
//        transMeta.addStep(stepMeta);
        return stepMeta;
    }

    @Override
    public StepMeta after(VisNode step, List<DatabaseMeta> databases, TransMeta transMeta, FlowConfig transGraph) throws Exception {
        UniqueNode uniqueNode =(UniqueNode)step;
        List<FlowHup> transHups = transGraph.getTransHups();
        List<FlowHup> newTransHups = new ArrayList<>();
        String sortStepName = step.getNodeName() + "-默认排序";
        for (FlowHup transHup : transHups) {

            if(transHup.getTo().equals(step.getNodeName())){
                newTransHups.add(new FlowHup(sortStepName,step.getNodeName()));
                newTransHups.add(new FlowHup(transHup.getFrom(), sortStepName));

                //处理pr节点是多路输出的情况
                dealMultiOut(transMeta, transGraph, sortStepName, transHup);
            }else{
                newTransHups.add(transHup);
            }
        }

        SortRowsStep sortRowsStep = new SortRowsStep();
        SortRowsNode visNode  = new SortRowsNode();
        visNode.setNodeName(sortStepName);

        List<SortField> collect = uniqueNode.getUniqueFields().stream().map(f -> new SortField(f, true)).collect(Collectors.toList());
        visNode.setSortFields(collect);
        StepMeta sortMeta  = sortRowsStep.decode(visNode, databases, transMeta, transGraph);
        sortMeta.setParentTransMeta(transMeta);
        transMeta.getSteps().add(sortMeta);

        List<TransHopMeta> hopMetas = newTransHups.stream().map(h -> new TransHopMeta(getStep(transMeta, h.getFrom()), getStep(transMeta, h.getTo()))).collect(Collectors.toList());
        transMeta.setTransHops(hopMetas);

        transGraph.setTransHups(hopMetas.stream().map(h -> new FlowHup(h.getFromStep().getName(), h.getToStep().getName())).collect(Collectors.toList()));
        return null;
    }
}
